package cn.ac.iie.ulss.indexer.worker;

import cn.ac.iie.common.util.avro.AvroUtils;
import cn.ac.iie.store.hbase.hbaseop.HbaseUtilOp;
import static cn.ac.iie.store.hbase.hbaseop.HbaseUtilOp.scanAllData;
import cn.ac.iie.ulss.indexer.runenvs.DataTypeMap;
import cn.ac.iie.ulss.indexer.runenvs.DefaultConstants;
import cn.ac.iie.ulss.indexer.runenvs.GlobalParas;
import cn.ac.iie.ulss.indexer.writer.DataSourceConfig;
import cn.ac.iie.ulss.indexer.writer.LuceneFileWriter;
import java.io.ByteArrayInputStream;
import java.io.File;
import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.security.MessageDigest;
import java.security.NoSuchAlgorithmException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import org.apache.avro.Protocol;
import org.apache.avro.Schema;
import org.apache.avro.generic.GenericArray;
import org.apache.avro.generic.GenericDatumReader;
import org.apache.avro.generic.GenericRecord;
import org.apache.avro.io.BinaryDecoder;
import org.apache.avro.io.DatumReader;
import org.apache.avro.io.DecoderFactory;
import org.apache.avro.util.Utf8;
import org.apache.hadoop.hive.metastore.api.FieldSchema;
import org.apache.log4j.Logger;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.DoubleField;
import org.apache.lucene.document.Field;
import org.apache.lucene.document.FieldType;
import org.apache.lucene.document.FloatField;
import org.apache.lucene.document.IntField;
import org.apache.lucene.document.LongField;
import org.apache.lucene.document.StringField;
import org.apache.lucene.document.TextField;
import org.apache.lucene.index.IndexWriter;

public class LuceneDataWriter implements Runnable {

    public static Logger log = Logger.getLogger(LuceneDataWriter.class.getName());
    private SimpleDateFormat secondFormat = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    public AtomicLong willwriteDocNum;
    public AtomicLong failwriteDocNum;
    String dbName;
    String tableName;
    public String tableType;
    String timeLable;
    long file_id = 0L;
    List<GenericRecord> innerDataBuf;
    int innerDataBufSize;
    volatile boolean isEnd;
    LuceneFileWriter normalLucWriter;
    AtomicBoolean isDiskBad;
    IndexWriter writer;
    private GenericArray arrayInt = null;
    private GenericArray arrayLong = null;
    private GenericArray arrayFloat = null;
    private GenericArray arrayDouble = null;
    private GenericArray arrayString = null;
    private GenericArray arrayBytes = null;
    private String hbaseRowKey = "";
    private int num_int_val = 0;
    private long num_long_val = 0L;
    private double num_double_val = 0.0D;
    private float num_float_val = 0.0F;
    private int arrayIdx = 0;
    private Integer dataTypeCase = Integer.valueOf(0);
    private String dataType = null;
    private String dataName = null;
    private String strData = null;
    private ByteBuffer bytesdata = null;
    StringBuilder arraySb = new StringBuilder("");
    private int hbaseWritenum = 0;
    private int luceneWritenum = 0;
    private long bg = System.currentTimeMillis();
    private LinkedHashMap<String, Field> idx2LuceneFieldMap = null;
    private LinkedHashMap<String, List<FieldSchema>> idx2colMap = null;
    private LinkedHashMap<String, FieldType> idx2LuceneFieldTypeMap = null;
    private ConcurrentHashMap<String, LinkedHashMap<String, List<FieldSchema>>> table2idx2colMap = null;
    private ConcurrentHashMap<String, LinkedHashMap<String, Field>> table2idx2LuceneFieldMap = null;
    private ConcurrentHashMap<String, LinkedHashMap<String, FieldType>> table2idx2LuceneFieldTypeMap = null;
    private Map<String, String> genKeyColumnMap = new HashMap<String, String>();
    public boolean isOuterIndexEable = false;
    public HashSet<String> outerIndexFieldsSet = new HashSet<String>();
    public ConcurrentHashMap<String, HashSet<Integer>> numberSets = new ConcurrentHashMap<String, HashSet<Integer>>();
    public ArrayList<String> hbaseKeyColumns = null;
    public ArrayList<String> hbaseStoreColumns = null;
    private Map<Utf8, Object> recordMap = new HashMap<Utf8, Object>();
    private HbaseUtilOp HbaseUtilOp = null;
    public static Random r = new Random(UUID.randomUUID().hashCode());
    Field field = null;
    GenericArray docSet = null;
    Iterator itor = null;
    ByteArrayInputStream dxxbis = null;
    BinaryDecoder dxxbd = null;
    Utf8 dataName1 = null;
    Object eveMapValue = null;

    LuceneDataWriter(DataSourceConfig ds, LuceneFileWriter norLucWriter, AtomicBoolean isdiskbad) {
        this.dbName = ds.getDbName();
        this.tableName = ds.getTbName();
        this.timeLable = ds.getTimeLable();
        this.innerDataBufSize = GlobalParas.writeInnerpoolBatchDrainSize;
        this.normalLucWriter = norLucWriter;
        this.isDiskBad = isdiskbad;
        this.innerDataBuf = new ArrayList(this.innerDataBufSize);
    }

    LuceneDataWriter() {
        this.innerDataBuf = new ArrayList(this.innerDataBufSize);
    }

    @Override
    public void run() {
        if (this.innerDataBuf.isEmpty()) {
            log.info("the write buffer size is 0,will return ...");
            return;
        }
        this.bg = System.currentTimeMillis();
        try {

            this.table2idx2colMap = ((ConcurrentHashMap) GlobalParas.tablename2table2idx2colMap.get(this.dbName + "." + this.tableName).take());
            this.table2idx2LuceneFieldMap = ((ConcurrentHashMap) GlobalParas.tablename2table2idx2LuceneFieldMap.get(this.dbName + "." + this.tableName).take());
            this.table2idx2LuceneFieldTypeMap = ((ConcurrentHashMap) GlobalParas.tablename2table2idx2LuceneFieldTypeMap.get(this.dbName + "." + this.tableName).take());
            this.idx2LuceneFieldMap = ((LinkedHashMap) this.table2idx2LuceneFieldMap.get(this.dbName + "." + this.tableName));
            this.idx2colMap = ((LinkedHashMap) this.table2idx2colMap.get(this.dbName + "." + this.tableName));
            this.idx2LuceneFieldTypeMap = ((LinkedHashMap) this.table2idx2LuceneFieldTypeMap.get(this.dbName + "." + this.tableName));
            Schema docschema = GlobalParas.utl_docschema;
            DatumReader<GenericRecord> docsingleDocReader = new GenericDatumReader<GenericRecord>(docschema);
            log.info("get cache index info use " + (System.currentTimeMillis() - this.bg) + " ms");
            Document doc = null;
            GenericRecord docsRecord = null;
            this.bg = System.currentTimeMillis();
            for (GenericRecord ms : this.innerDataBuf) {
                if (ms == null) {
                    log.error("the write not get null data！ it is wrong");
                } else if (ms != null) {
                    try {
                        docsRecord = ms;
                        docSet = (GenericArray) docsRecord.get(GlobalParas.docs_set);
                        itor = docSet.iterator();
                        while (itor.hasNext()) {
                            this.luceneWritenum++;
                            dxxbis = new ByteArrayInputStream(((ByteBuffer) itor.next()).array());
                            dxxbd = new DecoderFactory().binaryDecoder(dxxbis, null);
                            GenericRecord record = (GenericRecord) docsingleDocReader.read(null, dxxbd);

                            try {
                                doc = new Document();
                                for (String idxName : this.idx2LuceneFieldMap.keySet()) {
                                    //修改                                      
                                    this.dataType = ((FieldSchema) ((List) this.idx2colMap.get(idxName)).get(0)).getType();
                                    this.dataName = ((FieldSchema) ((List) this.idx2colMap.get(idxName)).get(0)).getName();
//                                        Object o = record.get(this.dataName);                                                                                                          
                                    if ("c_inputtime".equalsIgnoreCase(this.dataName)) {
                                        LongField f = new LongField("c_inputtime", System.currentTimeMillis() / 1000L, (FieldType) this.idx2LuceneFieldTypeMap.get("c_inputtime"));
                                        doc.add(f);
                                        continue;
                                    }
//                                        if (o == null) {
//                                            continue;
//                                        }
                                    dataName1 = new Utf8(this.dataName);
                                    eveMapValue = null;
                                    field = (Field) this.idx2LuceneFieldMap.get(idxName);
                                    this.dataTypeCase = (DataTypeMap.datatypeMap.get(this.dataType));
                                    if (this.dataTypeCase == null) {
                                        log.error("not support data type " + idxName + " " + this.dataType);
                                    } else {
                                        switch (this.dataTypeCase.intValue()) {
                                            case 0:
                                                try {
                                                    HashMap<Utf8, Object> intsMap = (HashMap<Utf8, Object>) record.get("int_s");
                                                    if (intsMap != null) {
                                                        if ((eveMapValue = intsMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);                                                           
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_int_val = Integer.parseInt(this.strData);
                                                                ((IntField) field).setIntValue(this.num_int_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 1:
                                                try {
                                                    HashMap<Utf8, Object> intaMap = (HashMap<Utf8, Object>) record.get("int_a");
                                                    if (intaMap != null) {
                                                        if ((eveMapValue = intaMap.get(dataName1)) != null) {
                                                            this.arrayInt = (GenericArray) eveMapValue;
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayInt.size(); this.arrayIdx += 1) {
                                                                IntField f = new IntField(idxName, -999, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayInt.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_int_val = Integer.parseInt(this.strData);
                                                                    f.setIntValue(this.num_int_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 2:
                                                try {
                                                    HashMap<Utf8, Object> longsMap = (HashMap<Utf8, Object>) record.get("long_s");
                                                    if (longsMap != null) {
                                                        if ((eveMapValue = longsMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_long_val = Long.parseLong(this.strData);
                                                                ((LongField) field).setLongValue(this.num_long_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 3:
                                                try {
                                                    HashMap<Utf8, Object> longaMap = (HashMap<Utf8, Object>) record.get("long_a");
                                                    if (longaMap != null) {
                                                        if ((eveMapValue = longaMap.get(dataName1)) != null) {
                                                            this.arrayLong = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayLong.size(); this.arrayIdx += 1) {
                                                                LongField f = new LongField(idxName, -999L, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayLong.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_long_val = Long.parseLong(this.arrayLong.get(this.arrayIdx).toString());
                                                                    f.setLongValue(this.num_long_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }

                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 4:
                                                try {
                                                    HashMap<Utf8, Object> longsMap = (HashMap<Utf8, Object>) record.get("long_s");
                                                    if (longsMap != null) {

                                                        if ((eveMapValue = longsMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_long_val = Long.parseLong(this.strData);
                                                                ((LongField) field).setLongValue(this.num_long_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 5:
                                                try {
                                                    HashMap<Utf8, Object> longaMap = (HashMap<Utf8, Object>) record.get("long_a");
                                                    if (longaMap != null) {
                                                        if ((eveMapValue = longaMap.get(dataName1)) != null) {
                                                            this.arrayLong = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayLong.size(); this.arrayIdx += 1) {
                                                                LongField f = new LongField(idxName, -999L, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayLong.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_long_val = Long.parseLong(this.arrayLong.get(this.arrayIdx).toString());
                                                                    f.setLongValue(this.num_long_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 6:
                                                try {
                                                    HashMap<Utf8, Object> longsMap = (HashMap<Utf8, Object>) record.get("long_s");
                                                    if (longsMap != null) {
                                                        if ((eveMapValue = longsMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_long_val = Long.parseLong(this.strData);
                                                                ((LongField) field).setLongValue(this.num_long_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 7:
                                                try {
                                                    HashMap<Utf8, Object> longaMap = (HashMap<Utf8, Object>) record.get("long_a");
                                                    if (longaMap != null) {
                                                        if ((eveMapValue = longaMap.get(dataName1)) != null) {
                                                            this.arrayLong = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayLong.size(); this.arrayIdx += 1) {
                                                                LongField f = new LongField(idxName, -999L, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayLong.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_long_val = Long.parseLong(this.arrayLong.get(this.arrayIdx).toString());
                                                                    f.setLongValue(this.num_long_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 8:
                                                try {
                                                    HashMap<Utf8, Object> floatsMap = (HashMap<Utf8, Object>) record.get("float_s");
                                                    if (floatsMap != null) {
                                                        if ((eveMapValue = floatsMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_float_val = Float.parseFloat(this.strData);
                                                                ((LongField) field).setLongValue(this.num_long_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 9:
                                                try {
                                                    HashMap<Utf8, Object> floataMap = (HashMap<Utf8, Object>) record.get("float_a");
                                                    if (floataMap != null) {
                                                        if ((eveMapValue = floataMap.get(dataName1)) != null) {
                                                            this.arrayFloat = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayFloat.size(); this.arrayIdx += 1) {
                                                                this.strData = this.arrayFloat.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_float_val = Float.parseFloat(this.strData);
                                                                    ((FloatField) field).setFloatValue(this.num_float_val);
                                                                    doc.add(field);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                            case 10:
                                                try {
                                                    HashMap<Utf8, Object> doublesMap = (HashMap<Utf8, Object>) record.get("double_s");
                                                    if (doublesMap != null) {
                                                        if ((eveMapValue = doublesMap.get(dataName1)) != null) {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_double_val = Double.parseDouble(this.strData);
                                                                ((DoubleField) field).setDoubleValue(this.num_double_val);
                                                                doc.add(field);
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(idxName + " " + e, e);
                                                }
                                                break;
                                            case 11:
                                                try {
                                                    HashMap<Utf8, Object> doubleaMap = (HashMap<Utf8, Object>) record.get("double_a");
                                                    if (doubleaMap != null) {
                                                        if ((eveMapValue = doubleaMap.get(dataName1)) != null) {
                                                            this.arrayDouble = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayDouble.size(); this.arrayIdx += 1) {
                                                                DoubleField f = new DoubleField(idxName, -999.0D, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayDouble.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_double_val = Double.parseDouble(this.arrayDouble.get(this.arrayIdx).toString());
                                                                    f.setDoubleValue(this.num_double_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 12:
                                                try {
                                                    HashMap<Utf8, Object> stringsMap = (HashMap<Utf8, Object>) record.get("string_s");
                                                    if (stringsMap != null) {
                                                        if ((eveMapValue = stringsMap.get(dataName1)) != null) {
                                                            if ((field instanceof StringField)) {
                                                                this.strData = eveMapValue.toString().trim();
                                                                //recordMap.put(dataName1, this.strData);
                                                                ((StringField) field).setStringValue(this.strData);
                                                                doc.add(field);
                                                            } else {
                                                                this.strData = eveMapValue.toString();
                                                                //recordMap.put(dataName1, this.strData);
                                                                ((TextField) field).setStringValue(this.strData);
                                                                doc.add(field);
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 13:
                                                try {
                                                    HashMap<Utf8, Object> stringaMap = (HashMap<Utf8, Object>) record.get("string_a");
                                                    if (stringaMap != null) {
                                                        if ((eveMapValue = stringaMap.get(dataName1)) != null) {
                                                            Utf8 utf8Value = (Utf8) eveMapValue;
                                                            this.arrayString = ((GenericArray) utf8Value);
                                                            if ((field instanceof StringField)) {
                                                                for (this.arrayIdx = 0; this.arrayIdx < this.arrayString.size(); this.arrayIdx += 1) {
                                                                    StringField f = new StringField(idxName, "", Field.Store.YES);
                                                                    f.setStringValue(this.arrayString.get(this.arrayIdx).toString());
                                                                    if (this.arrayIdx != 1) {
                                                                        arraySb.append("," + this.arrayString.get(this.arrayIdx).toString());
                                                                    } else {
                                                                        arraySb.append(this.arrayString.get(this.arrayIdx).toString());
                                                                    }
                                                                    doc.add(f);
                                                                }
                                                            } else {
                                                                for (this.arrayIdx = 0; this.arrayIdx < this.arrayString.size(); this.arrayIdx += 1) {
                                                                    TextField f = new TextField(idxName, "", Field.Store.YES);
                                                                    f.setStringValue(this.arrayString.get(this.arrayIdx).toString());
                                                                    if (this.arrayIdx != 1) {
                                                                        arraySb.append("," + this.arrayString.get(this.arrayIdx).toString());
                                                                    } else {
                                                                        arraySb.append(this.arrayString.get(this.arrayIdx).toString());
                                                                    }
                                                                    doc.add(f);
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 14:
                                                try {
                                                    HashMap<Utf8, Object> bytessMap = (HashMap<Utf8, Object>) record.get("bytes_s");
                                                    if (bytessMap != null) {
                                                        if ((eveMapValue = bytessMap.get(dataName1)) != null) {
                                                            this.bytesdata = ((ByteBuffer) eveMapValue);
                                                            //recordMap.put(dataName1, this.bytesdata.toString());
                                                            field.setBytesValue(this.bytesdata.array());
                                                            doc.add(field);
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 15:
                                                HashMap<Utf8, Object> bytesaMap = (HashMap<Utf8, Object>) record.get("bytes_a");
                                                if (bytesaMap != null) {
                                                    if ((eveMapValue = bytesaMap.get(dataName1)) != null) {
                                                        this.arrayBytes = ((GenericArray) eveMapValue);
                                                        try {
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayBytes.size(); this.arrayIdx += 1) {
                                                                this.bytesdata = ((ByteBuffer) this.arrayBytes.get(this.arrayIdx));
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.bytesdata.toString());
                                                                } else {
                                                                    arraySb.append(this.bytesdata.toString());
                                                                }
                                                                Field f = new Field(idxName, new byte[1], (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                f.setBytesValue(this.bytesdata.array());
                                                                doc.add(f);
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        } catch (Exception e) {
                                                            log.error(e, e);
                                                        }
                                                    }
                                                }
                                                break;
                                            case 16:
                                                HashMap<Utf8, Object> bytessMap = (HashMap<Utf8, Object>) record.get("bytes_s");
                                                if (bytessMap != null) {
                                                    if ((eveMapValue = bytessMap.get(dataName1)) != null) {
                                                        try {
                                                            this.strData = eveMapValue.toString();
                                                            //recordMap.put(dataName1, this.strData);
                                                            field.setStringValue(this.strData);
                                                            doc.add(field);

                                                        } catch (Exception e) {
                                                            log.error(e, e);
                                                        }
                                                    }
                                                }
                                                break;
                                            case 17:
                                                try {
                                                    HashMap<Utf8, Object> bytesaMap1 = (HashMap<Utf8, Object>) record.get("bytes_a");
                                                    if (bytesaMap1 != null) {
                                                        if ((eveMapValue = bytesaMap1.get(dataName1)) != null) {
                                                            this.arrayString = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayString.size(); this.arrayIdx += 1) {
                                                                this.bytesdata = ((ByteBuffer) this.arrayBytes.get(this.arrayIdx));
                                                                Field f = new Field(idxName, new byte[1], (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                f.setBytesValue(this.bytesdata.array());
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.bytesdata.toString());
                                                                } else {
                                                                    arraySb.append(this.bytesdata.toString());
                                                                }
                                                                doc.add(f);
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 18:
                                                HashMap<Utf8, Object> booleansMap = (HashMap<Utf8, Object>) record.get("boolean_s");
                                                if (booleansMap != null) {
                                                    if ((eveMapValue = booleansMap.get(dataName1)) != null) {
                                                        try {
                                                            this.strData = eveMapValue.toString();
                                                            //recordMap.put(dataName1, this.strData);
                                                            field.setStringValue(this.strData);
                                                            doc.add(field);
                                                        } catch (Exception e) {
                                                            log.error(e, e);
                                                        }
                                                    }
                                                }
                                                break;

                                            case 19:
                                                try {
                                                    HashMap<Utf8, Object> booleanaMap = (HashMap<Utf8, Object>) record.get("boolean_a");
                                                    if (booleanaMap != null) {
                                                        if ((eveMapValue = booleanaMap.get(dataName1)) != null) {
                                                            this.arrayString = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayString.size(); this.arrayIdx += 1) {
                                                                StringField f = new StringField(idxName, "", Field.Store.YES);
                                                                f.setStringValue(this.arrayString.get(this.arrayIdx).toString());
                                                                doc.add(f);
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.arrayString.get(this.arrayIdx).toString());
                                                                } else {
                                                                    arraySb.append(this.arrayString.get(this.arrayIdx).toString());
                                                                }
                                                            }
                                                            //recordMap.put(dataName1, arraySb.toString());
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            case 20:
                                                HashMap<Utf8, Object> tinysMap = (HashMap<Utf8, Object>) record.get("int_s");
                                                if (tinysMap != null) {
                                                    if ((eveMapValue = tinysMap.get(dataName1)) != null) {
                                                        try {
                                                            this.strData = eveMapValue.toString().trim();
                                                            //recordMap.put(dataName1, this.strData);
                                                            if (!"".equalsIgnoreCase(this.strData)) {
                                                                this.num_int_val = Integer.parseInt(this.strData);
                                                                ((IntField) field).setIntValue(this.num_int_val);
                                                                doc.add(field);
                                                            } else {
                                                                log.debug(idxName + " is blank or tab ");
                                                            }
                                                        } catch (Exception e) {
                                                            log.error(e, e);
                                                        }
                                                    }
                                                }
                                                break;
                                            case 21:
                                                try {
                                                    HashMap<Utf8, Object> tinyaMap = (HashMap<Utf8, Object>) record.get("int_a");
                                                    if (tinyaMap != null) {
                                                        if ((eveMapValue = tinyaMap.get(dataName1)) != null) {
                                                            this.arrayInt = ((GenericArray) eveMapValue);
                                                            for (this.arrayIdx = 0; this.arrayIdx < this.arrayInt.size(); this.arrayIdx += 1) {
                                                                IntField f = new IntField(idxName, -999, (FieldType) this.idx2LuceneFieldTypeMap.get(idxName));
                                                                this.strData = this.arrayInt.get(this.arrayIdx).toString().trim();
                                                                if (this.arrayIdx != 1) {
                                                                    arraySb.append("," + this.strData);
                                                                } else {
                                                                    arraySb.append(this.strData);
                                                                }
                                                                if (!"".equalsIgnoreCase(this.strData)) {
                                                                    this.num_int_val = Integer.parseInt(this.strData);
                                                                    f.setIntValue(this.num_int_val);
                                                                    doc.add(f);
                                                                } else {
                                                                    log.debug(idxName + " is blank or tab ");
                                                                }
                                                            }
                                                        }
                                                    }
                                                } catch (Exception e) {
                                                    log.error(e, e);
                                                }
                                                break;
                                            default:
                                                log.error("not support data type " + idxName + " " + this.dataType);
                                                break;
                                        }
                                        try {
                                            recordMap.put(dataName1, eveMapValue);
                                            if (GlobalParas.tableName2GenKey.get(this.dbName + "." + this.tableName).equalsIgnoreCase("true") && GlobalParas.hbaseKeyColumns.get(this.dbName + "." + this.tableName).contains(this.dataName)) {
                                                Object o = (Object) ("" + r.nextLong());
                                                hbaseRowKey += o.toString();
                                                recordMap.put(dataName1, o.toString());
                                            } else if (GlobalParas.tableName2GenKey.get(this.dbName + "." + this.tableName).equalsIgnoreCase("false") && GlobalParas.hbaseKeyColumns.get(this.dbName + "." + this.tableName).contains(this.dataName)) {
                                                hbaseRowKey += recordMap.get(dataName1).toString();
                                            }
                                            recordMap.put(dataName1, eveMapValue);
                                        } catch (NullPointerException e) {
                                            log.error(e, e);
                                        }

                                    }
                                }

                            } catch (Exception e) {
                                log.error(e, e);
                            }

                            /*
                             * lucene的数据写入完毕 就要向hbase写入数据，首先判断标的类型
                             * ----------------------- add to hbase code
                             * -----------------------
                             */
                            try {
                                if (this.tableType.equalsIgnoreCase(DefaultConstants.luceneHbaseTableType)) {
                                    if ("".equals(this.hbaseRowKey) || !this.HbaseUtilOp.getByRowkey(this.tableName, this.hbaseRowKey)) {
                                        StringField f = new StringField("key", "", Field.Store.YES);
                                        f.setStringValue(hbaseRowKey);
                                        doc.add(f);
                                    } else {
                                        doc = null;
                                    }
                                    this.HbaseUtilOp.hbasePutData(this.tableName, this.hbaseRowKey, DefaultConstants.lubaseDefaultFamily, this.hbaseStoreColumns, this.recordMap);
                                    this.hbaseWritenum++;
                                    this.hbaseRowKey = "";
                                    this.recordMap.clear();
                                }
                            } catch (NullPointerException e) {
                                log.debug(e, e);
                            }
                            /*
                             * ---------------- end of add to hbase code
                             * ------------ ****
                             */
                            try {
                                this.willwriteDocNum.incrementAndGet();
                                if (doc == null) {
                                    log.info("the doc is null,why ?");
                                }
                                if (this.writer == null) {
                                    log.info("the writer is null,why ?");
                                }
                                if (doc != null) {
                                    this.writer.addDocument(doc);
                                }
                            } catch (IOException e) {
                                this.isDiskBad.set(true);
                                log.error(e, e);
                                this.failwriteDocNum.incrementAndGet();
                            } catch (Exception e) {
                                log.error(e, e);
                                this.failwriteDocNum.incrementAndGet();
                            }

                        }
                    } catch (Exception e) {
                        log.error(e, e);
                    }
                }
            }
            log.info("write lucene file " + this.dbName + "." + this.tableName + " file_id:" + this.file_id + ": " + (System.currentTimeMillis() - this.bg) + " ms for " + this.luceneWritenum);
            log.info("write hbase " + this.dbName + "." + this.tableName + " use " + (System.currentTimeMillis() - this.bg) + " ms for " + this.hbaseWritenum);
            this.luceneWritenum = 0;
            this.hbaseWritenum = 0;
            this.innerDataBuf.clear();
        } catch (Exception ex) {
            log.error(ex, ex);
            this.innerDataBuf.clear();
        } finally {
            this.innerDataBuf.clear();
            try {
                GlobalParas.tablename2table2idx2LuceneFieldMap.get(this.dbName + "." + this.tableName).put(this.table2idx2LuceneFieldMap);
                GlobalParas.tablename2table2idx2colMap.get(this.dbName + "." + this.tableName).put(this.table2idx2colMap);
                GlobalParas.tablename2table2idx2LuceneFieldTypeMap.get(this.dbName + "." + this.tableName).put(this.table2idx2LuceneFieldTypeMap);
            } catch (Exception ex) {
                log.error(ex, ex);
            }
        }
    }

    public void setIsEnd(boolean isEnd) {
        this.isEnd = isEnd;
    }

    public boolean isIsEnd() {
        return this.isEnd;
    }

    public void addWriteDocNum(long inc) {
        this.willwriteDocNum.addAndGet(inc);
    }

    public long getWriteDocNum() {
        return this.willwriteDocNum.get();
    }

    public void clearWriteDocNum() {
        this.willwriteDocNum.set(0L);
    }

    public synchronized String getTimeLable() {
        return this.timeLable;
    }

    public synchronized void setTimeLable(String timeLable) {
        this.timeLable = timeLable;
    }

    public synchronized String getMD5(byte[] val) {
        MessageDigest digest = null;
        try {
            digest = MessageDigest.getInstance("MD5");
        } catch (NoSuchAlgorithmException ex) {
            log.error(ex, ex);
        }
        digest.reset();
        digest.update(val);
        BigInteger bigInt = new BigInteger(1, digest.digest());
        return bigInt.toString(16);
    }

    public static void main(String[] args) throws InterruptedException, IOException {
        AtomicLong al = new AtomicLong(0L);
        System.out.println(al.addAndGet(12L));
        System.out.println(al.get());
        Protocol docProtocol = Protocol.parse(new File("doc.json"));
        Schema docschema = docProtocol.getType("doc");    //doc_schame_name
        DatumReader<GenericRecord> docsingleDocReader = new GenericDatumReader<GenericRecord>(docschema);
        Map<String, Map<Utf8, Object>> map = new HashMap<String, Map<Utf8, Object>>(50, 0.8F);
        for (int i = 0; i < docschema.getFields().size(); i++) {
            String name = docschema.getFields().get(i).name();
            System.out.println(name);
//                                    Map<Utf8, Object> fieldSet = (HashMap<Utf8, Object>) record.get(name);
//                                    map.put(name, fieldSet);
        }
    }
}
